package org.study.hadoop.md.ch2;

import java.io.IOException;
import java.util.Map;

import org.apache.commons.lang.StringEscapeUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.study.hadoop.md.util.MRDPUtils;

/**
 * 倒排序索引概要1：给定一句用户评论信息，创建一个维基百科URL到一组回复帖子的ID的倒排索引。
 */
public class WikipediaIndex extends Configured implements Tool {
	
	public static class WikipediaReducer extends Reducer<Text, Text, Text, Text> {
		private Text result = new Text();
		
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			StringBuilder sb = new StringBuilder();
			for (Text id : values) {
				sb.append(id.toString() + " ");
			}
			result.set(sb.substring(0, sb.length() - 1).toString());
			context.write(key, result);
		}
	}
	
	public static class WikipediaMapper extends Mapper<Object, Text, Text, Text> {
		private Text link = new Text();
		private Text outKey = new Text();
		
		@Override
		protected void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			Map<String, String> parsed = MRDPUtils.transformXmlToMap(value.toString());
			String txt = parsed.get("Body");
			String postType = parsed.get("PostTypeId");
			String rowId = parsed.get("Id");
			if (txt == null || (postType != null && postType.equals("1"))) {
				return;
			}
			txt = StringEscapeUtils.unescapeHtml(txt.toLowerCase());
			link.set(getWikipediaURL(txt));
			outKey.set(rowId);
			context.write(link, outKey);
		}
	}
	
	public static String getWikipediaURL(String text) {
		int idx = text.indexOf("http://en.wikipedia.org");
		if (idx == -1) {
			return null;
		} else {
			return text;
		}
//		int idx_end = text.indexOf('"', idx + 1);
//		if (idx_end == -1) {
//			return null;
//		}
//		int idx_hash = text.indexOf('#', idx + 1);
//		if (idx_hash != -1 && idx_hash < idx_end) {
//			return text.substring(idx + 1, idx_hash);
//		} else {
//			return text.substring(idx + 1, idx_end);
//		}
	}

	@Override
	public int run(String[] args) throws Exception {
		Path outPath = new Path("out");
		Path inPath = new Path("post.txt");
		FileSystem fs = outPath.getFileSystem(getConf());
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		Job job = Job.getInstance(getConf());
		job.setJarByClass(WikipediaIndex.class);
		job.setMapperClass(WikipediaMapper.class);
		job.setCombinerClass(WikipediaReducer.class);
		job.setReducerClass(WikipediaReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		return job.waitForCompletion(true) ? 0 : 1;
	}
	
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new WikipediaIndex(), args);
		System.exit(res);
	}
}
